package cas.ihep.hadoop.rand;

import cas.ihep.hadoop.mc.YbjMCReducer3;
import cas.ihep.hadoop.util.Utils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class RandReducer extends Reducer<Text,Text,Text,Text> {

    private static final Log LOG = LogFactory.getLog(YbjMCReducer3.class);

    public void run(Context context)throws IOException {
        Configuration conf=context.getConfiguration();
        FileSystem fs=FileSystem.get(conf);
        String indir=conf.get(Utils.TmpInputDirKey);
        if(indir!=null && !indir.isEmpty()) {
            LOG.info("Deleting tmp input: "+indir);
            fs.delete(new Path(indir), true);
        }
        indir=conf.get(Utils.GlobalValueKey);
        if(indir!=null && !indir.isEmpty()){
            LOG.info("Deleting tmp global value: "+indir);
            fs.delete(new Path(indir),true);
        }
        indir=conf.get(Utils.RunfileUsetmp);
        if(indir!=null && !indir.isEmpty()){
            indir=conf.get(Utils.RunfileKey);
            if(indir!=null && !indir.isEmpty()){
                LOG.info("Deleting tmp runfile: "+indir);
                fs.delete(new Path(indir),true);
            }
        }
    }
}
